package atguigu.com.edu.app.dwd.db;


//支付成功表

import atguigu.com.edu.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.stream.Stream;

public class DwdTradePayDetailSuc {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(4);

         /*env.enableCheckpointing(50000L,CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000L);
        env.getCheckpointConfig().setCheckpointTimeout(6000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3,Time.days(30),Time.seconds(3)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop202:8020/edu/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //下单明细表
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
                "order_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "order_id string,\n" +
                "final_amount string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "od_ts string,\n" +
                "user_id string,\n" +
                "province_id string,\n" +
                "operate_date_id string,\n" +
                "operate_time string,\n" +
                "order_status string,\n" +
                "oi_ts string,\n" +
                "row_op_ts timestamp_ltz(3)\n" +

                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_detail", "dwd_trade_pay_detail_suc"));

            tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_pay_detail_suc_group"));


        // TODO 5. 筛选支付成功数据
        Table paymentInfo = tableEnv.sqlQuery("select\n" +
                        "data['user_id'] user_id,\n" +
                        "data['order_id'] order_id,\n" +
                        "data['payment_type'] payment_type,\n" +
                        "data['callback_time'] callback_time,\n" +
                        "`proc_time`,\n" +
                        "ts pi_ts\n"  +
                        "from topic_db\n" +
                        "where `table` = 'payment_info'\n"
             +
                  "and `type` = 'update' or `type` = 'insert' \n" +
                  "and data['payment_status']='1602'"
        );
        tableEnv.createTemporaryView("payment_info", paymentInfo);



        Table resultTable = tableEnv.sqlQuery("select \n" +
                "od.order_id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.final_amount,\n" +
                "od.date_id,\n" +
                "od.create_time,\n" +
                "od.od_ts,\n" +
                "od.user_id,\n" +
                "od.province_id,\n" +
                "od.order_status,\n" +
                "pi.payment_type,\n" +
                "pi.callback_time,\n" +


                "pi.pi_ts,\n" +
                "current_row_timestamp() row_op_ts\n" +
                "from dwd_trade_order_detail od \n" +
                "join payment_info pi\n" +
                "on od.order_id = pi.order_id" );
        tableEnv.createTemporaryView("result_table", resultTable);

        tableEnv.executeSql("create table dwd_trade_pay_detail_suc(\n" +
                "order_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "order_id string,\n" +
                "final_amount string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "od_ts string,\n" +
                "user_id string,\n" +
                "province_id string,\n" +
                "order_status string,\n" +
                "payment_type string,\n" +
                "callback_time string,\n" +


                "pi_ts string,\n" +
                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(order_id) not enforced\n" +
                ")"+ MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_suc"));

        tableEnv.executeSql("" +
                "insert into dwd_trade_pay_detail_suc select * from result_table");






    }


}
